package com.br.aiops.shuffle;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * kafka告警shuffle
 */
@Component
public class AlertLogKafkaShuffle {

    private static final Logger LOGGER = LoggerFactory.getLogger(AlertLogKafkaShuffle.class);

    public static final String FIELD_MESSAGE = "message";
    public static final String FIELD_SOURCE = "source";

    public static final String VALUE_NONE_MSG = "none-msg";
    public static final String VALUE_NONE_SERVICE = "none-service";
    public static final String VALUE_NONE_ERROR_CODE = "none-error-code";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${spring.kafka.consumer.partition-count}")
    private Integer partitionCount;

    @Value("${spring.kafka.consumer.topic}")
    private String topic;

    @KafkaListener(topics = "${spring.kafka.consumer.log-topic}")
    public void startConsume(ConsumerRecord<String, String> consumer) {
        String json = consumer.value();
        String key = resolve(json);
        int partition = Math.abs(key.hashCode()) % partitionCount;
        kafkaTemplate.send(topic, partition, key, json);
    }

    /**
     * 解析日志数据，并返回key
     * message示例：2021-09-23 09:56:48.286 [rule-service-6d8cb4776f-5gphg] [qtp247574430-1492] WARN [com.brcloud.rule.service.CachedRuleEngineServiceImpl] -ALERT_LEVEL:WARN ERROR_CODE:BR-LOG-10001 MESSAGE:responseTime:200 ,swift_number:1178
     * @param json
     * @return key
     */
    private String resolve(String json) {
        JSONObject jsonObject = JSONObject.parseObject(json);

        String message = jsonObject.getString(FIELD_MESSAGE);
        Matcher messageMatcher = matchMessage(message);
        String msg = Objects.nonNull(messageMatcher) ? messageMatcher.group(6) : VALUE_NONE_MSG;

        Matcher msgContentMatcher = matchMessageContent(msg);
        String errorCode = Objects.nonNull(msgContentMatcher) ? msgContentMatcher.group(2) : VALUE_NONE_ERROR_CODE;

        String source = jsonObject.getString(FIELD_SOURCE);
        Matcher sourceMatcher = matchSource(source);
        String service = Objects.nonNull(sourceMatcher) ? sourceMatcher.group(1) : VALUE_NONE_SERVICE;

        return service + "_" + errorCode;
    }

    private Pattern messagePattern = Pattern.compile("(.[0-9]+-[0-9]+-[0-9]+\\s[0-9]+:[0-9]+:[0-9]+.[0-9]+) \\[(.[\\w.-]+)\\] \\[(.[\\S]+)\\] (.[\\S]+)\\s+\\[(.[\\S]+)\\] -(.*)");
    private Pattern sourcePattern = Pattern.compile("/opt/logs/(.[a-z0-9A-Z-]+)/(.[a-z0-9A-Z-]+)/.*");
    private Pattern messageContentPattern = Pattern.compile("ALERT_LEVEL:(.[a-z0-9A-Z-]+) ERROR_CODE:(.[a-z0-9A-Z-]+) MESSAGE:(.*)");

    private Matcher matchSource(String source) {
        Matcher matcher = sourcePattern.matcher(source);
        if (matcher.find()) {
            return matcher;
        }
        return null;
    }

    private Matcher matchMessageContent(String msg) {
        Matcher matcher = messageContentPattern.matcher(msg);
        if (matcher.find()) {
            return matcher;
        }
        return null;
    }

    private Matcher matchMessage(String message) {
        Matcher matcher = messagePattern.matcher(message);
        if (matcher.find()) {
            return matcher;
        }
        return null;
    }
}
